package com.sui.bigdata.flink.sql.sink.message;

import com.alibaba.fastjson.JSONObject;
import com.feidee.fdcommon.model.StringResponse;
import com.feidee.fdcommon.util.HttpUtil;
import com.sui.bigdata.flink.sql.core.sink.MetricOutputFormat;
import com.sui.bigdata.flink.sql.sink.hbase.HbaseOutputFormat;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.calcite.shaded.com.google.common.collect.Maps;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.types.Row;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.StringEntity;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * @author YongChen
 * @date 2020/3/3 10:54
 * @description
 * @email yong_chen@sui.com
 */
public class MessageOutputFormat extends MetricOutputFormat {
    private static final Logger logger = LoggerFactory.getLogger(MessageOutputFormat.class);

    private static final String baseUrl = "http://search.feidee.com/fdalarm/alarm/sendStreamSetsAlarm.do?receiver=";

    private static final String pattern = "\\{[^}]*\\}";

    private String phone;

    private String template;

    protected String[] fieldNames;

    protected TypeInformation<?>[] fieldTypes;

    @Override
    public void configure(Configuration configuration) {

    }

    @Override
    public void open(int i, int i1) throws IOException {
        initMetric();
    }

    @Override
    public void writeRecord(Tuple2 tuple2) throws IOException {
        Tuple2<Boolean, Row> tupleTrans = tuple2;
        Boolean retract = tupleTrans.getField(0);
        if (!retract) {
            return;
        }
        Row record = tupleTrans.getField(1);
        if (record.getArity() != fieldNames.length) {
            return;
        }

        Pattern p = Pattern.compile(pattern);
        Matcher m = p.matcher(template);
        int i= 0;
        while (m.find()) {
            template = template.replace(m.group(),record.getField(i).toString());
            i++;
        }
        sendMsg(phone,template);
        outRecords.inc();

    }

    public  void sendMsg(String phone, String alarmInfo) {

        Map<String, String> body = Maps.newHashMap();
        body.put("text", alarmInfo +" at "+getCurrentTime());
        StringEntity stringEntity = new StringEntity(JSONObject.toJSONString(body), "UTF-8");
        stringEntity.setContentEncoding("UTF-8");
        stringEntity.setContentType("application/json");

        HttpPost httpPost = new HttpPost(baseUrl+phone);
        RequestConfig requestConfig =RequestConfig.custom().setConnectTimeout(30000).build();
        httpPost.setEntity(stringEntity);
        httpPost.setConfig(requestConfig);
        httpPost.setEntity(stringEntity);

        StringResponse response = null;
        try {
            response = HttpUtil.post(httpPost);
            if ("success".equals(response.getResponseBody())) {
                logger.info("send msg succeed:  "+ alarmInfo);
            }
        } catch (Exception e) {
            logger.error("send msg fail: " + alarmInfo );
            logger.error(" record: {}",alarmInfo);
            outFailRecords.inc();
        }
        outRecords.inc();
    }
    public static String getCurrentTime(){
        SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        return  df.format(new Date(System.currentTimeMillis()));
    }

    @Override
    public void close() throws IOException {

    }


    public static MessageOutputFormat.MessageOutputFormatBuilder buildMessageOutputFormat() {
        return new MessageOutputFormat.MessageOutputFormatBuilder();
    }

    public static class MessageOutputFormatBuilder {

        private final MessageOutputFormat messageOutputFormat;

        protected MessageOutputFormatBuilder() {
            this.messageOutputFormat = new MessageOutputFormat();
        }

        public MessageOutputFormat.MessageOutputFormatBuilder setPhone(String phone) {
            messageOutputFormat.phone = phone;
            return this;
        }

        public MessageOutputFormat.MessageOutputFormatBuilder setTemplate(String template) {
            messageOutputFormat.template = template;
            return this;
        }

        public MessageOutputFormat.MessageOutputFormatBuilder setFieldNames(String[] fieldNames) {
            messageOutputFormat.fieldNames = fieldNames;
            return this;
        }

        public MessageOutputFormat.MessageOutputFormatBuilder setFieldTypes(TypeInformation<?>[] fieldTypes) {
            messageOutputFormat.fieldTypes = fieldTypes;
            return this;
        }

        public MessageOutputFormat finish() {
            if (messageOutputFormat.phone == null) {
                throw new IllegalArgumentException("No phone supplied.");
            }

            if (messageOutputFormat.template == null) {
                throw new IllegalArgumentException("No template supplied.");
            }

            Matcher m = Pattern.compile(pattern).matcher(messageOutputFormat.template);

            List<String> fieldTemplate = new ArrayList();
            while (m.find() ) {
                String g = m.group();
                String target = g.substring(1, g.length() - 1);
                fieldTemplate.add(target);
            }

            if (!fieldTemplate.equals(Arrays.asList(messageOutputFormat.fieldNames))){
                throw new IllegalArgumentException("template can't match fieldNames.");
            }

            return messageOutputFormat;
        }
    }
}
